Skip to content
图编排引擎：LangGraph 和多 Agent 架构
复杂的 Agent 产品基本都是多 Agent 架构。为什么呢？
为什么要用多 Agent 架构
单 Agent 架构下，所有 tool 的描述、每个功能的 prompt 都放到 system prompt 里。实际上执行每个功能只需要其中一部分 prompt，但每次都全带上。
这样会导致 token 消耗更高，更重要的是很多无关信息干扰，思考效率低还更容易出错。
而如果你拆分成多个 Agent 呢？
- 每个 Agent 只保留需要的 prompt：执行功能的时候，消耗的 token 更少，没有无关信息干扰，准确率也更高
- 多 Agent 可以并行思考：单 Agent 只有一个大脑，需要一步步思考、调用 tool；而多 Agent 的多个大脑可以并行思考，主 Agent 下发任务，子 Agent 并行处理完成后返回
- 多角色互相讨论、纠错：单 Agent 虽然可以加上反思阶段，但相当于自己给自己纠错；而多 Agent 每个都是不同的角色，可以互相讨论纠错
基于这三个原因：
- 决策准确率更高、token 消耗更低：每个 Agent 只带必要的最少 prompt，没有冗余信息干扰，虽然调用 LLM 次数多了，但更省 token、决策更准、更稳定
- 并行思考和任务处理：主管分派任务，子 Agent 并行处理，整体效率更高
- 多角色互相讨论，纠错能力更强：可以互相监督、互相纠错，比单个 Agent 自己反思更靠谱，复杂任务表现更强
现在复杂 Agent 产品基本都是多 Agent 架构的。
实现 Multi Agent 就需要学习 LangGraph 了。用到的 api 还是 LangChain 那些，但它多了一套图编排引擎。我们学了 LangChain 的组件，学了 LCEL 的线性编排，今天来学一下 LangGraph 的图编排引擎。
先看下它们在整个 LangChain 生态里的位置：
LangChain 的体系可以分三层：
- 组件层（LangChain Core）：LLM 接口、Document Loaders、Text Splitters、Embeddings、Vector Stores、Output Parsers——稳定的基础设施，与编排框架无关
- 编排层：左边是 LangChain Chains / LCEL（线性编排 A→B→C，简单场景适用）；右边是 LangGraph（图编排，复杂 Agent 适用）
- 应用层：左边用 LangChain 做文档问答、简单 RAG；右边用 LangGraph 做多步 Agent、人机协作流程、多 Agent 系统
所以两套编排不是替代关系，是互补：简单线性用 LangChain，需要循环 / 多 Agent / 复杂决策就用 LangGraph。
什么时候用 LangChain，什么时候用 LangGraph？简单线性流程 → LangChain（流程走完就结束，没有循环）；有循环、多 Agent、工具自动重试 → LangGraph。比如 RAG 检索到的问题要让大模型评估下，不行就重新生成问题、重新检索——这就是循环，是 agentic RAG，需要 LangGraph 来做。
我们直接通过代码来学一下：
bash
mkdir langgraph-test
cd langgraph-test
npm init -y安装依赖：
bash
pnpm install @langchain/langgraph @langchain/core @langchain/openai dotenv zod创建 .env：
env
OPENAI_API_KEY=sk-xxx
OPENAI_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1
MODEL_NAME=qwen-plus第一个图：StateGraph
创建 src/basic-graph.mjs：
javascript
import { Annotation, END, START, StateGraph } from "@langchain/langgraph";
const StateAnnotation = Annotation.Root({
text: Annotation({
reducer: (_prev, next) => next,
default: () => "",
}),
});
const step1 = (state) => ({ text: `${state.text} -> step1` });
const step2 = (state) => ({ text: `${state.text} -> step2` });
const graph = new StateGraph(StateAnnotation)
.addNode("step1", step1)
.addNode("step2", step2)
.addEdge(START, "step1")
.addEdge("step1", "step2")
.addEdge("step2", END)
.compile();
// 导出为 Mermaid：可复制到 https://mermaid.live 或 Markdown 的 ```mermaid 代码块
const drawable = await graph.getGraphAsync();
const mermaid = drawable.drawMermaid({ withStyles: true });
console.log(mermaid);
const result = await graph.invoke({ text: "hello" });
console.log("result:", result);创建 StateGraph 图，添加两个节点（node），加上固定的 START、END 节点，然后用边（edge）连起来，编译后执行。
Annotation 用于创建 State，指定默认值（default）和合并逻辑（reducer）。这样我们基于 LangGraph 的第一个图就完成了。
图中当然有分支和循环。先试一下分支：
分支：addConditionalEdges
src/conditional-routing.mjs：
javascript
import { Annotation, END, START, StateGraph } from "@langchain/langgraph";
const StateAnnotation = Annotation.Root({
query: Annotation({
reducer: (_prev, next) => next,
default: () => "",
}),
route: Annotation({
reducer: (_prev, next) => next,
default: () => "chat",
}),
answer: Annotation({
reducer: (_prev, next) => next,
default: () => "",
}),
});
const router = (state) => {
const isMath = /[+\-*/]/.test(state.query);
return { route: isMath ? "math" : "chat" };
};
const mathNode = (state) => {
try {
return { answer: String(eval(state.query)) };
} catch {
return { answer: "表达式无法计算" };
}
};
const chatNode = (state) => ({ answer: `你说的是：${state.query}` });
const graph = new StateGraph(StateAnnotation)
.addNode("router", router)
.addNode("math", mathNode)
.addNode("chat", chatNode)
.addEdge(START, "router")
.addConditionalEdges("router", (state) => state.route, {
math: "math",
chat: "chat",
})
.addEdge("math", END)
.addEdge("chat", END)
.compile();
// 导出为 Mermaid
const drawable = await graph.getGraphAsync();
const mermaid = drawable.drawMermaid({ withStyles: true });
console.log(mermaid);
console.log("result:", await graph.invoke({ query: "你好" }));
console.log("result:", await graph.invoke({ query: "10 * 8" }));用 addConditionalEdges 添加分支：判断文本如果有 +-*/ 字符就走 math 分支，否则走 chat 分支。
循环：也是用分支实现
src/loop-retry.mjs：
javascript
import { Annotation, END, START, StateGraph } from "@langchain/langgraph";
const StateAnnotation = Annotation.Root({
tries: Annotation({
reducer: (_prev, next) => next,
default: () => 0,
}),
ok: Annotation({
reducer: (_prev, next) => next,
default: () => false,
}),
message: Annotation({
reducer: (_prev, next) => next,
default: () => "",
}),
});
const attempt = (state) => {
const tries = state.tries + 1;
const ok = tries >= 3;
return {
tries,
ok,
message: ok ? `第 ${tries} 次成功` : `第 ${tries} 次失败，继续重试`,
};
};
const graph = new StateGraph(StateAnnotation)
.addNode("attempt", attempt)
.addEdge(START, "attempt")
.addConditionalEdges("attempt", (state) => (state.ok ? "done" : "retry"), {
retry: "attempt",
done: END,
})
.compile();
// 导出为 Mermaid
const drawable = await graph.getGraphAsync();
const mermaid = drawable.drawMermaid({ withStyles: true });
console.log(mermaid);
const result = await graph.invoke({ tries: 0 });
console.log("result:", result);同样用 addConditionalEdges 判断条件满足就到 END 节点，否则重新路由到之前的节点，这样就可以实现循环效果（重试 3 次）。
Checkpointer：保存执行状态
经过这几个例子，应该能看出节点之间是怎么通信的：通过 state。那把 state 保存下来不就是把当前图的执行状态保存下来了么？
这个通过 CheckpointerSaver 的 api 就可以保存。创建 src/checkpointer-memory.mjs：
javascript
import {
Annotation,
END,
MemorySaver,
START,
StateGraph,
} from "@langchain/langgraph";
const StateAnnotation = Annotation.Root({
visitCount: Annotation({
reducer: (_prev, next) => next,
default: () => 0,
}),
message: Annotation({
reducer: (_prev, next) => next,
default: () => "",
}),
});
/** 每跑一轮图，给「当前会话」访问次数 +1 */
function recordVisit(state) {
const visitCount = state.visitCount + 1;
const message =
visitCount === 1
? "这是你在本会话里第 1 次进入。"
: `这是你在本会话里第 ${visitCount} 次进入`;
return { visitCount, message };
}
const graph = new StateGraph(StateAnnotation)
.addNode("recordVisit", recordVisit)
.addEdge(START, "recordVisit")
.addEdge("recordVisit", END);
const checkpointer = new MemorySaver();
const app = graph.compile({ checkpointer });
const user1 = { configurable: { thread_id: "用户-小张" } };
const user2 = { configurable: { thread_id: "用户-小李" } };
const res1 = await app.invoke({}, user1);
const res2 = await app.invoke({}, user1);
const res3 = await app.invoke({}, user1);
const res4 = await app.invoke({}, user2);
console.log(res1);
console.log(res2);
console.log(res3);
console.log(res4);我们用 MemorySaver 来把 state 保存到内存里，这样下次就会基于上次的 state 继续执行（同一 thread_id 会话内 visitCount 累加，不同 thread_id 隔离）。当然，还可以保存到 sqlite、redis 等，分别用 SqliteSaver、RedisSaver 等 api。
interrupt：图的打断与恢复
我们用 Cursor 之类的 coding agent，它经常会让你确认，确认后再继续执行，这种打断功能咋做呢？LangGraph 提供了 interrupt 的 api。
创建 src/graph-interrupt.mjs：
javascript
import { createInterface } from "node:readline/promises";
import {
Annotation,
Command,
END,
MemorySaver,
START,
StateGraph,
interrupt,
} from "@langchain/langgraph";
const StateAnnotation = Annotation.Root({
actionSummary: Annotation({
reducer: (_prev, next) => next,
default: () => "",
}),
userInput: Annotation({
reducer: (_prev, next) => next,
default: () => "",
}),
});
/** 展示一笔待确认的转账 */
const showTransfer = () => ({
actionSummary: "向张三转账 ¥100（模拟，不会真扣款）",
});
/** 停在这里等人输入；resume 的值会写进 userInput */
const waitConfirm = (state) => {
const text = interrupt({
hint: "终端里输入「确认」或备注后回车，图才会继续",
actionSummary: state.actionSummary,
});
return { userInput: String(text) };
};
const graph = new StateGraph(StateAnnotation)
.addNode("showTransfer", showTransfer)
.addNode("waitConfirm", waitConfirm)
.addEdge(START, "showTransfer")
.addEdge("showTransfer", "waitConfirm")
.addEdge("waitConfirm", END)
.compile({ checkpointer: new MemorySaver() });
// 导出为 Mermaid
const drawable = await graph.getGraphAsync();
const mermaid = drawable.drawMermaid({ withStyles: true });
console.log(mermaid);
const config = { configurable: { thread_id: "interrupt-demo" } };
const paused = await graph.invoke({}, config);
console.log("\n待你确认：", paused.__interrupt__?.[0]?.value);
const rl = createInterface({ input: process.stdin, output: process.stdout });
const line = (await rl.question("> ")).trim();
await rl.close();
if (!line) {
console.error("未输入，退出。");
process.exit(1);
}
const done = await graph.invoke(new Command({ resume: line }), config);
console.log("结果：", done);用 interrupt 中断图的执行，等待用户输入之后再次 invoke，传入 new Command({ resume: 'xxx' })，这样图就会在上次断点位置继续执行。这里用了 nodejs 的 readline 包读取键盘输入。这样就可以实现图的中断、恢复了。
Prebuilt：内置节点
此外，有些常用的节点，langgraph 给封装好了，放到 prebuilt 下。比如我们要调用 tool，用 graph 的写法怎么写呢？
创建 model 的节点、tool 的节点，然后加一个 conditional 节点，判断如果有 tool call 就走 tool 节点，否则走 END。但不用自己写，langgraph 内置了 ToolNode 和 toolsCondition 的 api：
src/prebuilt-tool-node.mjs：
javascript
import "dotenv/config";
import { HumanMessage } from "@langchain/core/messages";
import { tool } from "@langchain/core/tools";
import {
END,
MessagesAnnotation,
START,
StateGraph,
} from "@langchain/langgraph";
import { ToolNode, toolsCondition } from "@langchain/langgraph/prebuilt";
import { ChatOpenAI } from "@langchain/openai";
import { z } from "zod";
import { getProductBySku } from "./inventory-mock.mjs";
const getProductStock = tool(
async ({ sku }) => getProductBySku(sku),
{
name: "get_product_stock",
description: "按 SKU 查商品名与库存，SKU 如 SKU-001。",
schema: z.object({
sku: z.string().describe("商品 SKU"),
}),
}
);
const tools = [getProductStock];
const llm = new ChatOpenAI({
modelName: process.env.MODEL_NAME,
apiKey: process.env.OPENAI_API_KEY,
configuration: {
baseURL: process.env.OPENAI_BASE_URL,
},
}).bindTools(tools);
async function agent(state) {
const response = await llm.invoke(state.messages);
return { messages: response };
}
const toolNode = new ToolNode(tools);
const graph = new StateGraph(MessagesAnnotation)
.addNode("agent", agent)
.addNode("tools", toolNode)
.addEdge(START, "agent")
.addConditionalEdges("agent", toolsCondition, ["tools", END])
.addEdge("tools", "agent")
.compile();
const result = await graph.invoke({
messages: [
new HumanMessage(
"查一下 SKU-001 的库存还有多少，回答里带上商品名和数字。"
),
],
});
// 导出为 Mermaid
const drawable = await graph.getGraphAsync();
const mermaid = drawable.drawMermaid({ withStyles: true });
console.log(mermaid);
const last = result.messages.at(-1);
console.log(last?.content ?? result.messages);用到的 inventory-mock.mjs（假数据，模拟按 SKU 查库存接口）：
javascript
const rows = [
{ sku: "SKU-001", name: "无线鼠标", stock: 42 },
{ sku: "SKU-002", name: "机械键盘", stock: 7 },
{ sku: "SKU-003", name: "USB-C 线缆", stock: 120 },
];
export function getProductBySku(sku) {
const key = String(sku).trim().toUpperCase();
const row = rows.find((r) => r.sku.toUpperCase() === key);
if (!row) return JSON.stringify({ found: false, sku: String(sku).trim() });
return JSON.stringify({ found: true, ...row });
}当然，像这么常用的 agent loop 自然也给封装好了，就是 createAgent 的 api：
prebuilt-agent.mjs：
javascript
import "dotenv/config";
import { HumanMessage } from "@langchain/core/messages";
import { ChatOpenAI } from "@langchain/openai";
import { MemorySaver } from "@langchain/langgraph";
import { createAgent, tool } from "langchain";
import { z } from "zod";
import { getProductBySku } from "./inventory-mock.mjs";
const getProductStock = tool(
async ({ sku }) => getProductBySku(sku),
{
name: "get_product_stock",
description: "按 SKU 查商品名与库存，SKU 如 SKU-001。",
schema: z.object({
sku: z.string().describe("商品 SKU"),
}),
}
);
const model = new ChatOpenAI({
modelName: process.env.MODEL_NAME,
apiKey: process.env.OPENAI_API_KEY,
configuration: {
baseURL: process.env.OPENAI_BASE_URL,
},
});
const agent = createAgent({
model,
tools: [getProductStock],
systemPrompt: "你是仓库助手。问库存时必须调用 get_product_stock（模拟数据），禁止编造。",
checkpointer: new MemorySaver(),
});
const result = await agent.invoke(
{ messages: [new HumanMessage("SKU-002 还剩多少库存？")] },
{ configurable: { thread_id: "demo-thread" } }
);
// 导出为 Mermaid
const drawable = await agent.graph.getGraphAsync();
const mermaid = drawable.drawMermaid({ withStyles: true });
console.log(mermaid);
const last = result.messages.at(-1);
console.log(last?.content ?? result);直接用它来跑 agent loop，看它的图——和刚才手写的图一模一样，这个 api 内部就是基于 LangGraph 构建的 agent loop 的图。
多 Agent：Supervisor-Worker 模式
学完 LangGraph 的图，我们来写一个多 Agent 的架构。多 Agent 最常用的是 Supervisor-Worker 模式，也就是"主管-工人"模式。
langchain 提供了这种多 Agent 架构的包 @langchain/langgraph-supervisor：
bash
pnpm install @langchain/langgraph-supervisor创建 multi-agent-supervisor.mjs：
javascript
import "dotenv/config";
import { HumanMessage } from "@langchain/core/messages";
import { createSupervisor } from "@langchain/langgraph-supervisor";
import { ChatOpenAI } from "@langchain/openai";
import { createAgent, tool } from "langchain";
import { z } from "zod";
import { lookupCityTrivia, lookupWeather } from "./simple-mock.mjs";
const model = new ChatOpenAI({
modelName: process.env.MODEL_NAME,
apiKey: process.env.OPENAI_API_KEY,
configuration: {
baseURL: process.env.OPENAI_BASE_URL,
},
});
const lookupWeatherTool = tool(
async ({ city }) => lookupWeather(city),
{
name: "lookup_weather",
description: "查询某城市当日天气概况（气温区间、天气、空气质量等）。",
schema: z.object({
city: z.string().describe("城市名，如 杭州"),
}),
}
);
const lookupCityTriviaTool = tool(
async ({ city }) => lookupCityTrivia(city),
{
name: "lookup_city_trivia",
description: "查询与某城市相关的一句趣味知识。",
schema: z.object({
city: z.string().describe("城市名，如 杭州"),
}),
}
);
/** 子代理 A：只回答「天气」类问题 */
const weatherAgent = createAgent({
name: "weather_agent",
description: "专门查天气",
model,
tools: [lookupWeatherTool],
systemPrompt: "你只处理天气。用户提到城市时，用 lookup_weather 查询后再用中文简短说明。",
});
/** 子代理 B：只回答「城市小知识」 */
const triviaAgent = createAgent({
name: "trivia_agent",
description: "专门讲与城市相关的小知识；必须调用 lookup_city_trivia。",
model,
tools: [lookupCityTriviaTool],
systemPrompt: "你只讲城市小知识。先 lookup_city_trivia，再用人话转述，不要编造工具里没有的内容。",
});
/**
* Supervisor：根据用户问的是「天气」还是「小知识」切换子代理。
* （真实业务里还可以再加更多子代理，思路一样。）
*/
const workflow = createSupervisor({
agents: [weatherAgent.graph, triviaAgent.graph],
llm: model,
prompt: `你是调度员，只负责选人，不要自己报气温、也不要自己讲城市百科。
- 问天气、气温、下不下雨、空气 → 用 weather_agent
- 问小知识、名胜、历史、一句介绍 → 用 trivia_agent
`,
});
const app = workflow.compile();
const drawable = await app.getGraphAsync();
console.log(drawable.drawMermaid({ withStyles: true }));
const input = {
messages: [
new HumanMessage("查一下杭州的天气，再讲一条和杭州有关的小知识。"),
],
};
const nodePath = [];
let finalState = null;
const stream = await app.stream(input, { streamMode: ["updates", "values"] });
for await (const event of stream) {
const [mode, payload] = event;
if (mode === "updates" && payload && typeof payload === "object") {
nodePath.push(...Object.keys(payload));
} else if (mode === "values") {
finalState = payload;
}
}
console.log("路径:", nodePath.join(" → "));
const last = finalState?.messages?.at(-1);
console.log(last?.content ?? finalState?.messages);我们用 createAgent 创建了 2 个子 Agent（一个查天气，一个查城市历史），然后用 createSupervisor 创建主管 Agent。
用 stream 可以拿到整个图运行过程的状态。它的 state 内容挺多的，所以支持几种模式：
updates是增量模式，过滤出这个节点增量修改的 statevalues是全量模式，给你所有的 state
我们这里用 updates 模式拿到经过的节点的名字，最后的回复用 values 模式拿。
用到的查询代码实现（simple-mock.mjs，假接口演示 supervisor 如何把问题分给不同子代理）：
javascript
function normCity(city) {
return String(city).trim();
}
const weatherTable = {
杭州: { summary: "多云转小雨", tempHighC: 22, tempLowC: 15, aqi: "良" },
北京: { summary: "晴", tempHighC: 26, tempLowC: 12, aqi: "轻度污染" },
上海: { summary: "阴", tempHighC: 20, tempLowC: 16, aqi: "良" },
};
const triviaTable = {
杭州: "西湖文化景观是世界文化遗产之一。",
北京: "故宫是世界上现存规模最大的古代宫殿建筑群之一。",
上海: "外滩万国建筑博览群是近代城市历史的缩影。",
};
/** 查某地当日天气摘要（模拟） */
export function lookupWeather(city) {
const c = normCity(city);
const w = weatherTable[c];
if (!w) {
return JSON.stringify({
city: c,
summary: "暂无该城市数据，以下为占位",
tempHighC: 20,
tempLowC: 12,
aqi: "—",
});
}
return JSON.stringify({ city: c, ...w });
}
/** 查与某城市相关的一句小知识（模拟） */
export function lookupCityTrivia(city) {
const c = normCity(city);
const line = triviaTable[c];
return JSON.stringify({
city: c,
trivia: line ?? `没有为「${c}」准备内置小知识，可换杭州 / 北京 / 上海试试。`,
});
}跑一下，这样我们第一个多 Agent 的代码就跑通了。输入"查一下杭州的天气，再讲一条和杭州有关的小知识"，可以看到路径大致是 supervisor → weather_agent → supervisor → trivia_agent → supervisor。
调试多 Agent
虽然用 stream 的 values 模式可以打印 state，但是它内容太多了。如果想看一下执行过程，最好的方式是断点调试。
通过调试，就可以清晰的看到整个 graph 的流转过程。也可以看到 @langchain/langgraph-supervisor 的多 Agent 架构的实现原理，就是在 state 里保存了 messages 数组来传递信息（子 agent 回答结束，会跟着两条假消息：AIMessage(name: weather_agent, content: transferring back to supervisor) 和 ToolMessage(name: transfer_back_to_supervisor)，这是 supervisor 包的内部实现）。
我们学 LangChain 的组件层花了比较多时间，学编排层的 LCEL、LangGraph 都是很快的，一两节搞定。
代码上传了课程仓库：https://github.com/QuarkGluonPlasma/ai-agent-course-code
总结
这节我们学了 LangGraph 和多 Agent 架构。
我们理清了 3 个用多 Agent 架构的理由：
- prompt 拆分到多个 Agent 中去，更纯净，token 消耗少，不容易决策出错
- 多个 Agent 可以并行思考和执行任务
- 多个 Agent 基于各自的角色可以相互讨论、纠错
复杂的 Agent 产品基本都是多 Agent 架构。
我们学了 LangGraph 的图怎么创建：
- state 用
Annotation创建，包括 default（默认值）、reducer（值怎么合并） - 图用
StateGraph创建，可以添加 node（节点）、edge（边） - 边可以用
addConditionalEdges添加路由分支，基于这个也可以实现循环 - 可以用
MemorySaver等 checkpointer 保存节点的 state，这样就可以恢复上次执行状态了 - 用
interrupt可以做图执行过程的打断，之后再次 invoke 传入 resume Command 即可恢复执行 - 还学了 prebuilt 的
ToolNode、toolsCondition以及createAgent这些内置的节点、图
学完 LangGraph 的图之后，我们学了多 Agent：
- 多 Agent 一般是 Supervisor-Worker 的架构
- 直接用
@langchain/langgraph-supervisor这个包就行，它封装了这套架构 - 用
stream可以看到图执行过程中的 state，分别用 updates、values 可以增量、全量看到节点输出的 state - 打印太多的话可以直接用断点调试来看多 Agent 的流转过程
- Supervisor 主管节点只负责任务分发，Worker 来做具体的任务执行
后面我们的项目实战都是基于这种多 Agent 的架构来写。
踩坑提示：
- 子 Agent 的 prompt 要收紧，严格只回答自己的领域。不然模型可能"越权"，比如 DeepSeek 在 weather_agent 里把天气和城市知识两个问题都答了，supervisor 看到已回答就不再调 trivia_agent。
- 如果报
400 An assistant message with 'tool_calls' must be followed by tool messages，通常是模型兼容性问题，换个模型或升级依赖版本。- createSupervisor 的子 Agent 逻辑上可以并行，但实际执行中肯定有一个先跑的先后顺序。
预览到此为止，输入密码解锁全文
解锁后本机会记住，同密码的其他文章也无需重复输入